/*
 * (c) Copyright 2009 Hewlett-Packard Development Company, LP
 * All rights reserved.
 * [See end of file]
 */

package com.hp.hpl.jena.clusteredtdb.pgraph;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import lib.Tuple;

import com.hp.hpl.jena.clusteredtdb.base.BulkTupleLib;
import com.hp.hpl.jena.graph.Node;
import com.hp.hpl.jena.graph.Triple;
import com.hp.hpl.jena.graph.TripleMatch;
import com.hp.hpl.jena.tdb.nodetable.NodeTable;
import com.hp.hpl.jena.tdb.store.GraphTriplesTDB;
import com.hp.hpl.jena.tdb.store.NodeId;
import com.hp.hpl.jena.tdb.store.TripleTable;
import com.hp.hpl.jena.util.iterator.ExtendedIterator;
import com.hp.hpl.jena.util.iterator.NiceIterator;

public class NetworkedGraphTDB extends GraphTriplesTDB {
		
	ArrayList<Triple> addBuffer;
	HashSet<Node> addNodeBuffer;
	long lastAddTime;
	long totalTime;
	long totsize = 0;
	
	//protected NetworkedGraphTDB() {}   // Must call init!

	public NetworkedGraphTDB(TripleTable tripleTable, NodeTable nodeTable) {
	    
	    super(tripleTable, null, /* Prefixes,*/ null, /*reorderTransform,*/ null /*location*/) ;
		this.addBuffer = new ArrayList<Triple>(ClusteredTDB.writebufsize);
		this.addNodeBuffer = new HashSet<Node>(ClusteredTDB.writebufsize/5);
		lastAddTime = System.currentTimeMillis();
		totalTime = lastAddTime;
		
	}

	@Override
	protected ExtendedIterator graphBaseFind(TripleMatch m)
	{
		Node s = m.getMatchSubject() ;
		Node p = m.getMatchPredicate() ;
		Node o = m.getMatchObject() ;

		NodeId subj = idForNode(s) ;
		if ( subj == NodeId.NodeDoesNotExist )
			return new com.hp.hpl.jena.util.iterator.NullIterator() ;

		NodeId pred = idForNode(p) ;
		if ( pred == NodeId.NodeDoesNotExist )
			return new com.hp.hpl.jena.util.iterator.NullIterator() ;

		NodeId obj = idForNode(o) ;
		if ( obj == NodeId.NodeDoesNotExist )
			return new com.hp.hpl.jena.util.iterator.NullIterator() ;

//		if ( subj < 0 && subj != NodeTable.NodeNotConcrete )
//	throw new JenaException("Subject error") ;
//if ( pred < 0 && pred != NodeTable.NodeNotConcrete )
//		throw new JenaException("Predicate error") ;
//		if ( obj < 0 && obj != NodeTable.NodeNotConcrete )
//		throw new JenaException("Object error") ;
		
		boolean s_set = ( subj != NodeId.NodeIdAny ) ;
		boolean p_set = ( pred != NodeId.NodeIdAny ) ;
		boolean o_set = ( obj  != NodeId.NodeIdAny ) ;
		
		if ( s_set && p_set && o_set )
		{
			// s p o
			if( getNodeTupleTable().find(subj, pred, obj).hasNext() )
				return new com.hp.hpl.jena.util.iterator.SingletonIterator(new Triple(s,p,o)) ;
			else
				return new com.hp.hpl.jena.util.iterator.NullIterator() ;
		}

		Iterator<Tuple<NodeId>> _iter = getNodeTupleTable().find(subj, pred, obj) ;
	    //@SuppressWarnings("unchecked")
		Iterator<Triple> iter = BulkTupleLib.convertToTriples((BulkNodeTable)this.getNodeTupleTable().getNodeTable(), _iter, ClusteredTDB.bufsize) ;
		return new MapperIterator(iter) ;
    }
	
	@Override
	public void performAdd( Triple t ) {
		addBuffer.add(t);
		addNodeBuffer.add(t.getSubject());
		addNodeBuffer.add(t.getPredicate());
		addNodeBuffer.add(t.getObject());
	}
	
	@Override
	public void sync(boolean force) {
		this.flushAddBuffer();
		
		////commented debug alternative: add one by one
		/*Iterator<Triple> it = addBuffer.iterator();
		while(it.hasNext()) {
			Triple t = it.next();
			super.performAdd(t);
		}
		addBuffer.clear();*/

		// No prefices => crash.
		try { super.sync(force); } catch (Exception ex) {}
	}
	
	//TODO Handle duplicates correctly!!
	protected void flushAddBuffer() {
		int size = this.addBuffer.size();
		Map<Node,NodeId> nodeIdNodeMap = this.storeNodes(this.addNodeBuffer);
		ArrayList<Tuple<NodeId>> triples = new ArrayList<Tuple<NodeId>>(addBuffer.size());
		this.addNodeBuffer.clear();
		//System.out.println("ADDBUFSIZE: "+ addBuffer.size());
		Iterator<Triple> it = addBuffer.iterator();
		while(it.hasNext()) {
			Triple t = it.next();
			//System.out.println("ADDBUFSIZE2: "+ addBuffer.size());
			//System.out.println("TRIP: "+ t.getSubject() + " "+ t.getPredicate() +" " + t.getObject());
			NodeId[] triple = new NodeId[3];
			triple[0] = nodeIdNodeMap.get(t.getSubject());
			triple[1] = nodeIdNodeMap.get(t.getPredicate());
			triple[2] = nodeIdNodeMap.get(t.getObject());			
			triples.add(Tuple.create(triple));
		}
		
		if ( ! this.getIndexSPO().bulkAdd(triples) )
			return;

		if ( getIndexPOS() != null )
			getIndexPOS().bulkAdd(triples);

		if ( this.getIndexOSP() != null )
			getIndexOSP().bulkAdd(triples);
		
		addBuffer.clear();
		
		totsize += size;
		
		System.out.println(size + " triples added in " +(System.currentTimeMillis() - lastAddTime)+"ms.  Total this session: " +totsize +" triples in "+ (System.currentTimeMillis() - totalTime) + "ms");
		lastAddTime = System.currentTimeMillis();
		
	}
	
	private BulkTupleIndex getIndexSPO() { return (BulkTupleIndex)getNodeTupleTable().getTupleTable().getIndex(0) ; }
	private BulkTupleIndex getIndexPOS() { return (BulkTupleIndex)getNodeTupleTable().getTupleTable().getIndex(1) ; }
	private BulkTupleIndex getIndexOSP() { return (BulkTupleIndex)getNodeTupleTable().getTupleTable().getIndex(2) ; }
	
	protected Map<Node,NodeId> storeNodes(Set<Node> nodes) {
		return ((BulkNodeTable)this.getNodeTupleTable().getNodeTable()).storeNodes(nodes);
	}

	private final NodeId idForNode(Node node)
    {
        if ( node == null || node == Node.ANY )
            return NodeId.NodeIdAny ;
        return this.getNodeTupleTable().getNodeTable().getNodeIdForNode(node) ;
    }
	
	static class MapperIterator extends NiceIterator
	{
	    Iterator<Triple> iter ;
	    MapperIterator(Iterator<Triple> iter)
	    {
	        this.iter = iter ;
	    }
	    
	    @Override
	    public boolean hasNext() { return iter.hasNext() ; } 
	    
	    @Override
	    public Triple next() { return iter.next(); }
	}
	
}

/*
 * (c) Copyright 2009 Hewlett-Packard Development Company, LP
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. The name of the author may not be used to endorse or promote products
 *    derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
